#include <librdkafka/rdkafkacpp.h>

#include <iostream>
#include <string>
#include <vector>

class ConsumerCb : public RdKafka::RebalanceCb, public RdKafka::ConsumeCb
{
   public:
    void rebalance_cb(RdKafka::KafkaConsumer *consumer, RdKafka::ErrorCode err, std::vector<RdKafka::TopicPartition *> &partitions) override
    {
        if (err == RdKafka::ERR__ASSIGN_PARTITIONS)
        {
            std::cout << "Assigning partitions..." << std::endl;
            consumer->assign(partitions);
        }
        else if (err == RdKafka::ERR__REVOKE_PARTITIONS)
        {
            std::cout << "Revoking partitions..." << std::endl;
            consumer->unassign();
        }
        else
        {
            std::cerr << "Rebalance error: " << RdKafka::err2str(err) << std::endl;
            consumer->unassign();
        }
    }

    void consume_cb(RdKafka::Message &message, void *opaque) override
    {
        if (message.err())
        {
            std::cerr << "Consumer error: " << message.errstr() << std::endl;
        }
        else
        {
            std::cout << "Received message: " << static_cast<const char *>(message.payload()) << std::endl;
        }
    }
};

int main()
{
    std::string brokers = "localhost:9092";
    std::string group_id = "test_group";
    std::string topic = "test_topic";
    std::string errstr;

    // 创建配置对象
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

    // 设置broker列表和group ID
    conf->set("metadata.broker.list", brokers, errstr);
    conf->set("group.id", group_id, errstr);
    conf->set("enable.auto.commit", "true", errstr);

    // 创建消费者回调
    ConsumerCb consumer_cb;
    conf->set("rebalance_cb", static_cast<RdKafka::RebalanceCb *>(&consumer_cb), errstr);
    conf->set("consume_cb", static_cast<RdKafka::ConsumeCb *>(&consumer_cb), errstr);

    // 创建Kafka消费者
    RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);
    if (!consumer)
    {
        std::cerr << "Failed to create consumer: " << errstr << std::endl;
        return 1;
    }

    // 订阅主题
    RdKafka::ErrorCode err = consumer->subscribe({topic});
    if (err)
    {
        std::cerr << "Failed to subscribe to topic: " << RdKafka::err2str(err) << std::endl;
        delete consumer;
        delete conf;
        delete tconf;
        return 1;
    }

    // 消费消息
    while (true)
    {
        RdKafka::Message *msg = consumer->consume(1000);
        consumer_cb.consume_cb(*msg, nullptr);
        delete msg;
    }

    // 取消订阅并清理资源
    consumer->unsubscribe();
    delete consumer;
    delete conf;
    delete tconf;

    return 0;
}
